package net.techfuser.channelorderreport;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;
import java.util.List;

class ChannelOrderMySqlSink extends RichSinkFunction<List<ChannelOrder>> {

    private Connection connection = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://qnh-flink-data.mysql.rds.aliyuncs.com:3306/flink_data_output?useUnicode=true&characterEncoding=utf-8";
        String user = "qnh_flink";
        String passwd = "Md7fpbWSHpBa";
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, passwd);
    }

    public void upsert(ChannelOrder row) throws SQLException {
        String upsert = "INSERT INTO channel_order_summary(" +
                "ent_id,  region_code, region_name, channel_keyword, date, " +
                "qnh_valid_order_num, qnh_gmv_value, qnh_total_sale_value, qnh_logistics_value, qnh_package_fee, " +
                "complate_order_num, channel_valid_order_num, channel_gmv_value) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE " +
                "qnh_valid_order_num = qnh_valid_order_num + ?, " +
                "qnh_gmv_value = qnh_gmv_value + ?, " +
                "qnh_total_sale_value = qnh_total_sale_value + ?, " +
                "qnh_logistics_value = qnh_logistics_value + ?, " +
                "qnh_package_fee = qnh_package_fee + ?, " +

                "complate_order_num = complate_order_num + ?, " +
                "channel_valid_order_num = channel_valid_order_num + ?, " +
                "channel_gmv_value = channel_gmv_value + ?, " +
                "last_modify_time = ?";

        ps = connection.prepareStatement(upsert);
        ps.setLong(1, row.ent_id);
        ps.setString(2, row.region_code);
        ps.setString(3, row.region_name);
        ps.setString(4, row.channel_keyword);
        ps.setDate(5, new Date(row.date.getTime()));

        ps.setInt(6, row.qnh_valid_order_num);
        ps.setDouble(7, row.qnh_gmv_value);
        ps.setDouble(8, row.qnh_total_sale_value);
        ps.setDouble(9, row.qnh_logistics_value);
        ps.setDouble(10, row.qnh_package_fee);

        ps.setInt(11, row.complate_order_num);
        ps.setInt(12, row.channel_valid_order_num);
        ps.setDouble(13, row.channel_gmv_value);


        ps.setInt(14, row.qnh_valid_order_num);
        ps.setDouble(15, row.qnh_gmv_value);
        ps.setDouble(16, row.qnh_total_sale_value);
        ps.setDouble(17, row.qnh_logistics_value);
        ps.setDouble(18, row.qnh_package_fee);

        ps.setInt(19, row.complate_order_num);
        ps.setInt(20, row.channel_valid_order_num);
        ps.setDouble(21, row.channel_gmv_value);
        ps.setTimestamp(22, new Timestamp(System.currentTimeMillis()));

        ps.execute();
    }

    @Override
    public void invoke(List<ChannelOrder> rows, Context context) throws Exception {
        for(ChannelOrder row : rows){
            upsert(row);
        }
    }

    @Override
    public void close() throws Exception {
        if(null != connection) connection.close();
        if(null != ps) ps.close();
    }

}